package com.theaf.core.mq.service;

import com.aliyun.openservices.ons.api.*;
import com.aliyun.openservices.ons.api.exception.ONSClientException;
import com.theaf.core.mq.message.MessageBody;
import com.theaf.core.mq.properties.ConsumerProperties;
import com.theaf.core.utils.MethodHelper;
import com.theaf.core.utils.SerializationUtils;
import com.theaf.core.utils.SpringUtils;
import com.theaf.core.utils.StringUtils;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import java.io.IOException;
import java.lang.reflect.Field;
import java.lang.reflect.Method;
import java.util.Arrays;
import java.util.Properties;
/**
 * 生产者服务类
 *
 * @author luojuan
 * @create 2017-11-28 9:49
 **/
public class ConsumerServer {
    private static final Log logger = LogFactory
            .getLog(ConsumerServer.class);

    public static void start(String topic , String tags , ConsumerProperties consumerProperties){
        logger.info("Aliyun message queue consumer starts up." + "TOPIC=" + topic + " || TAGS=" + tags);
        Class propertiesClass =  consumerProperties.getClass();
        Field[] fs = propertiesClass.getDeclaredFields();
        Properties properties = new Properties();
        Arrays.asList(fs).forEach(ele -> {
            ele.setAccessible(true);
            try {
                Object val = ele.get(consumerProperties);
                properties.put( StringUtils.toUpperFristChar(ele.getName()) , val);
            } catch (IllegalAccessException e) {
                logger.info("The aliyun message queue consumer startup failed.");
                throw new ONSClientException("MQMessage failed" , e);
            }
        });
        Consumer consumer = ONSFactory.createConsumer(properties);
        consumer.subscribe(topic, tags, (Message message, ConsumeContext context) ->{
            try {
                MessageBody messageBody = (MessageBody) SerializationUtils.deserialize(message.getBody());
                String beanName = messageBody.getBeanName();
                Object[] param = messageBody.getParam();
                String methodName = messageBody.getMethodName();
                Object o = SpringUtils.getBean(beanName);
                if( o == null ) {
                    logger.error("消费失败,找不到接口:" + beanName);
                    return Action.CommitMessage;
                }
                if( param == null )
                {
                    try {
                        Method m = MethodHelper.getMethod(o.getClass(), methodName);
                        if( m == null )
                        {
                            logger.error("消费失败,找不到方法:" + beanName + "#" + methodName);
                            throw new ONSClientException("消费失败,找不到方法:" + beanName + "#" + methodName);
                        }
                        m.invoke(o);
                    } catch (Exception e) {
                        throw new ONSClientException("Consumption failed" , e);
                    }
                }else{
                    try {
                        Method  m = MethodHelper.getMethod(o.getClass(), methodName,param);
                        if( m == null )
                        {
                            logger.error("消费失败,找不到方法:" + beanName + "#" + methodName);
                            throw new ONSClientException("消费失败,找不到方法:" + beanName + "#" + methodName);
                        }
                       m.invoke(o, param);

                    } catch (Exception e) {
                        throw new ONSClientException("Consumption failed" , e);
                    }
                }
            } catch (IOException e) {
                e.printStackTrace();
                logger.error("消费失败!");
                throw new ONSClientException("Consumption failed" , e);
            } catch (Exception e){
                e.printStackTrace();
                logger.error("消费失败!" + message);
                throw new ONSClientException("Consumption failed" , e);
            }
            return Action.CommitMessage;
        });
        consumer.start();
        logger.info("Aliyun message queue consumer started successfully.");
    }

}
